package com.scala.weblog

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.ProcessingTime

object StructuredStreaming {
    //定义数据库表对应实体类
    case class Weblog(datatime:String,
                      userid:String,
                      searchname:String,
                      retorder:String,
                      cliorder:String,
                      cliurl:String)
    def main(args: Array[String]): Unit = {

       /* val spark  = SparkSession.builder()
            .master("local[*]")
            .appName("streaming").getOrCreate()

        val df = spark
            .readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", "node2:9092")
            .option("subscribe", "weblogs")
            .load()

        import spark.implicits._
        val lines = df.selectExpr("CAST(value AS STRING)").as[String]
        val wordCounts=lines.flatMap(_.split(" ")).groupBy("value").count()

        val url ="jdbc:mysql://node3:3306/test"
        val username="root"
        val password="123456"


        val query = wordCounts.writeStream
            .outputMode("update")
            .format("console")

            .start()
        query.awaitTermination()*/



        val spark  = SparkSession.builder()
            .master("local[2]")
            .appName("streaming").getOrCreate()

        val df = spark
            .readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", "node2:9092,node3:9092,node4:9092")
            .option("subscribe", "weblogs")
            .load()
        import spark.implicits._
        val lines = df.selectExpr("CAST(value AS STRING)").as[String]
        val weblog = lines.map(_.split(","))
            .map(x => Weblog(x(0),x(1), x(2),x(3),x(4),x(5)))
        val titleCount = weblog
            .groupBy("searchname").count().toDF("titleName","count")

        val url ="jdbc:mysql://node3:3306/test"
        val username="root"
        val password="123"

        val writer = new JDBCSink(url,username,password)
        val query = titleCount.writeStream
            .foreach(writer)
            .outputMode("update")
            .trigger(ProcessingTime("25 seconds"))
            .start()
        query.awaitTermination()
    }


}
